package com.zte.claa.inficombo.csa.app.service.dataredeal.impl;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.mina.util.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.zte.claa.inficombo.csa.app.config.DataCombineProps;
import com.zte.claa.inficombo.csa.app.model.dev.DevTypeModel;
import com.zte.claa.inficombo.csa.app.model.redis.DevMeteData;
import com.zte.claa.inficombo.csa.app.service.csa.impl.BaseCacheManager;
import com.zte.claa.inficombo.csa.app.service.dataredeal.IMetedataRedealer;
import com.zte.claa.inficombo.csa.app.service.dataredeal.MetedataRedealService;
import com.zte.claa.inficombo.csa.app.service.persistent.CSADataPersistService;
import com.zte.claa.inficombo.csa.app.service.redis.CSAMessagePublisher;
import com.zte.claa.inficombo.csa.app.util.JsonUtils;

/**
 * metedata二次合并服务，主要针对北秦和新山鹰的电气火灾设备，拆分指标分批上报的情况。
 * 
 * @author 0283000115
 *
 */
@Service
public class MetedataCombineService implements IMetedataRedealer, InitializingBean, DisposableBean {
	// 日志
	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
	@Autowired
	private DataCombineProps combinePara;
	
	@Autowired
	private BaseCacheManager cacheManager;
	
	@Autowired
	private CSADataPersistService persisService;
	
	@Autowired
	private MetedataRedealService redealService;
	
	// 消息发布器
	private CSAMessagePublisher msgPublisher = null;
	
	// 待合并设备类型集
	private Set<String> devtypeSet = null;
	
	// 涉及合并的deveui集合
	private Set<String> deveuiSet = null;
	
	// 合并补发检测定时器（针对异常情况，处理超时未合并的数据）
	private ScheduledExecutorService expDataCleanScheduler = null;
	
	@Override
	public void afterPropertiesSet() throws Exception {
		this.deveuiSet = new ConcurrentHashSet<String>();
		// 设置需要合并的设备类型.
		this.devtypeSet = new HashSet<String>();
		String s = this.combinePara.getDevtypes();
		if(s != null && !s.trim().equals("")){
			String[] devtypeIdArr = s.trim().split(",");
			for(String devtypeId : devtypeIdArr){
				if(devtypeId != null && !devtypeId.trim().equals("")){
					this.devtypeSet.add(devtypeId.trim());
					// 向二次处理入口服务注册.
					this.redealService.registMetedataRedealer(devtypeId, this);
				}
			}
		}
		// 启动合并补发检查定时器. 
		this.expDataCleanScheduler = Executors.newScheduledThreadPool(1);
		//分包数据过期清理.
		Runnable task = new Runnable(){
			@Override
			public void run() {
				checkCombineData();
			}
		};
		// 每隔10秒钟调度一次.
		this.expDataCleanScheduler.scheduleAtFixedRate(task, 10, this.combinePara.getTimeoutcheckperiod(), TimeUnit.SECONDS);
		logger.info("[**]Combine timer check task is submitted and will be scheduled in 10 seconds later!");	
	}

	@Override
	public void destroy() throws Exception {
		this.expDataCleanScheduler.shutdown();
		this.expDataCleanScheduler = null;
		this.devtypeSet.clear();
		this.devtypeSet = null;
		this.msgPublisher = null;
	}

	@Override 
	public void setCSAMessagePublisher(CSAMessagePublisher publisher) {
		this.msgPublisher = publisher;
	}

	@Override
	public void submitMetedata(DevMeteData data) {
		// 检查是否需要合并；检查是否需要发布；
		try {
			String devEui = data.getDeveui();
			DevMeteData curData = this.cacheManager.getCombineData(devEui);
			if(curData == null){
				this.deveuiSet.add(devEui);
				this.cacheManager.putCombineData(data);
				return;
			}
			Map<String, Object> curMap = curData.getMdata();
			if(curMap == null){
				curMap = new HashMap<String, Object>();
				curData.setMdata(curMap);
			}
			curMap.putAll(data.getMdata());
			if(this.combinePara.isSendafter1stcombine()){
				this.cacheManager.removeCombineData(devEui);
				this.deveuiSet.remove(devEui);
				this.processCombineData(curData);				
			} else if(this.isTimeout(curData.getRpttime())){
				this.cacheManager.removeCombineData(devEui);
				this.deveuiSet.remove(devEui);
				this.processCombineData(curData);				
			} else {
				this.cacheManager.putCombineData(curData);
				this.deveuiSet.add(devEui);
			}			
		}catch(Exception e){
			logger.error("submitMetedata failed! "+data.getDeveui()+", err="+e.getMessage());
		}
	}
	
	// 提交一批待合并数据（前提是需要合并）
	public void submitMetedatas(List<DevMeteData> dataList) {
		if(dataList == null || dataList.size() == 0){
			return;
		}
		for(DevMeteData data : dataList){
			this.submitMetedata(data);
		}
	}
	
	// 判断数据合并是否已经超时.
	private boolean isTimeout(Timestamp dataOrgRptTime){
		return (System.currentTimeMillis() - dataOrgRptTime.getTime()) > (this.combinePara.getMaxdatainterval()*1000);
	}
	
	/**
	 * 定时检查待合并数据（是否超时）
	 */
	private void checkCombineData(){
		// 复制一份deveui列表，避免和删除操作冲突.
		List<String> tmpList = new ArrayList<String>();	
		tmpList.addAll(this.deveuiSet);
		for(String devEui : tmpList){
			try {
				DevMeteData data = this.cacheManager.getCombineData(devEui);
				if (data == null) {
					continue;
				}
				if (this.isTimeout(data.getRpttime())) {
					this.cacheManager.removeCombineData(devEui);
					this.deveuiSet.remove(devEui);
					this.processCombineData(data);
				}
			}catch(Exception e){
				logger.error("checkCombineData failed: deveui="+devEui+", err="+e.getMessage());
			}
		}	
		tmpList.clear();
		tmpList = null;
	}
	
	// 持久化并发布数据.
	private void processCombineData(DevMeteData data) throws Exception {
		String devTypeId = data.getDevtype();
		DevTypeModel dtm = this.cacheManager.getDevTypeModel(devTypeId);
		// 1.持久化数据.
		this.persisService.saveDevMeteData(data, dtm.getBase().getTbname());
		// 2.构造广播数据，进行redis发布...
		String pubMsg = JsonUtils.toJson(data);
		this.msgPublisher.publishMessage(pubMsg);
		logger.info("[*^_^*]Combine and publish metedata message ok: "+pubMsg);
	}

}
